package com.fy.dao.template;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zhshuo
 * @date 2021/12/28 13:58
 */
@Component
public class HbaseTemplate {

    @Autowired
    private Connection connection;

    private final String encoding = "utf-8";

    public static final TableName TABLE_MESSAGE = TableName.valueOf("message:message");

    /**
     * 根据rowKey 获取单行数据
     * @param rowKey 指定 rowKey
     * @return {@link Result}
     * @throws IOException
     */
    public Result getRow(final String rowKey) throws IOException {
        Get get = new Get(rowKey.getBytes(encoding));
        return getTable().get(get);
    }

    /**
     * 根据正则表达式批量获取rowKey
     * @param regxKey 正则表达式的 rowKey
     * @return {@link List<Result>}
     * @throws IOException
     */
    public List<Result> getRegexRow(final String regxKey) throws IOException {
        List<Result> list = new ArrayList<>();
        RegexStringComparator rc = new RegexStringComparator(regxKey);
        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, rc);
        Scan scan = new Scan();
        scan.setFilter(rowFilter);
        ResultScanner rs = getTable().getScanner(scan);
        for (Result result : rs) {
            list.add(result);
        }
        return list;
    }

    /**
     * 根据regxKey正则匹配数据,取出num条
     * @param regxKey
     * @param num
     * @return
     * @throws IOException
     */
    public List<Result> getRegexRow( final String regxKey, final int num) throws IOException {
        List<Result> list = new ArrayList<>();

        FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        RegexStringComparator rc = new RegexStringComparator(regxKey);
        RowFilter rf = new RowFilter(CompareOperator.EQUAL, rc);
        if (num > 0) {
            Filter filterNum = new PageFilter(num);
            fl.addFilter(filterNum);
        }
        fl.addFilter(rf);
        Scan scan = new Scan();
        scan.setFilter(fl);
        ResultScanner rscanner = getTable().getScanner(scan);
        for (Result result : rscanner) {
            list.add(result);
        }
        return list;
    }

    /**
     * 根据指定rowKey的start到end的范围 查询
     * @param startKey 开始rowKey
     * @param stopKey 结束rowKey
     * @return {@link List<Result>}
     * @throws IOException
     */
    public List<Result> getStartRowAndEndRow( final String startKey,
                                             final String stopKey) throws IOException {
        List<Result> list = new ArrayList<>();

        Scan scan = new Scan();
        scan.withStartRow(startKey.getBytes(encoding));
        scan.withStopRow(stopKey.getBytes(encoding));
        ResultScanner rscanner = getTable().getScanner(scan);
        for (Result result : rscanner) {
            list.add(result);
        }
        return list;
    }

    /**
     * 根据正则表达式的指定rowKey过滤,并且指定rowkey范围
     * @param startKey 开始rowKey
     * @param stopKey 结束rowKey
     * @param regxKey 正则表达式rowKey
     * @return
     * @throws IOException
     */
    public List<Result> getRegexRow(final String startKey,
                                    final String stopKey, final String regxKey) throws IOException {
        List<Result> list = new ArrayList<>();
        RegexStringComparator rc = new RegexStringComparator(regxKey);
        RowFilter rf = new RowFilter(CompareOperator.EQUAL, rc);
        Scan scan = new Scan();
        scan.withStartRow(startKey.getBytes(encoding));
        scan.withStopRow(stopKey.getBytes(encoding));
        scan.setFilter(rf);
        ResultScanner rscanner = getTable().getScanner(scan);
        for (Result result : rscanner) {
            list.add(result);
        }
        return list;
    }

    /**
     * 确定startKey和endKey的范围，根据regKey匹配数据,取出num条
     * @param startKey
     * @param stopKey
     * @param regxKey
     * @param num
     * @return
     * @throws IOException
     */
    public List<Result> getRegexRow(final String startKey,
                                    final String stopKey, final String regxKey, final int num) throws IOException {
        List<Result> list = new ArrayList<>();
        FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        RegexStringComparator rc = new RegexStringComparator(regxKey);
        RowFilter rf = new RowFilter(CompareOperator.EQUAL, rc);
        if (num > 0) {
            Filter filterNum = new PageFilter(num);
            fl.addFilter(filterNum);
        }
        fl.addFilter(rf);
        Scan scan = new Scan();
        scan.withStartRow(startKey.getBytes(encoding));
        scan.withStopRow(stopKey.getBytes(encoding));
        scan.setFilter(fl);
        ResultScanner rscanner = getTable().getScanner(scan);
        for (Result result : rscanner) {
            list.add(result);
        }
        return list;
    }

    /**
     * 添加数据
     */
    public void addData(String rowKey,String family, String[] column, String[] value) throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));
        for (int j = 0; j < column.length; j++) {
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column[j]),Bytes.toBytes(value[j]));
        }
        getTable().put(put);
    }

    /**
     * 删除记录
     */
    public void delRecord(String... rowKeys) throws IOException {
        List<Delete> list = new ArrayList<>();
        for (String rowKey : rowKeys) {
            Delete del = new Delete(Bytes.toBytes(rowKey));
            list.add(del);
        }
        getTable().delete(list);
    }

    /**
     * 修改一条数据
     */
    public void updateTable( String rowKey, String familyName, String column[],
                     String value[]) throws IOException{
        Put put = new Put(Bytes.toBytes(rowKey));
        for (int j = 0; j < column.length; j++) {
            put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column[j]),
                    Bytes.toBytes(value[j]));
        }
        getTable().put(put);
    }

    /**
     * 查找最新的一条数据,或者说倒序查询
     */
    public Result getNewRow() throws IOException {
        Filter filterNum = new PageFilter(1);// 每页展示条数
        Scan scan = new Scan();
        scan.setFilter(filterNum);
        scan.setReversed(true);
        ResultScanner scanner = getTable().getScanner(scan);
        return scanner.next();
    }

    /**
     * 指定rowKey的正则,查找最新的一条数据,或者说倒序查询
     *
     */
    public Result getNewRow( final String regxKey) throws IOException {
        FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        RegexStringComparator rc = new RegexStringComparator(regxKey);
        RowFilter rf = new RowFilter(CompareOperator.EQUAL, rc);
        Filter filterNum = new PageFilter(1);// 每页展示条数
        fl.addFilter(rf);
        fl.addFilter(filterNum);
        Scan scan = new Scan();
        scan.setFilter(fl);
        scan.setReversed(true);
        ResultScanner scanner = getTable().getScanner(scan);
        return scanner.next();
    }

    /**
     * 正则查出所有匹配的key
     */
    public List<String> queryKeys(String regxKey) throws IOException {
        List<String> list = new ArrayList<>();
        PrefixFilter filter = new PrefixFilter(regxKey.getBytes(encoding));
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner scanner = getTable().getScanner(scan);
        for (Result rs : scanner) {
            list.add(new String(rs.getRow()));
        }
        return list;
    }

    /**
     * 增加表中对应字段的值
     */
    public long incrQualifier(String tableName, String cf, String rowKey, String column, long num) throws IOException {
        long qualifie = getTable()
                .incrementColumnValue(rowKey.getBytes(encoding), cf.getBytes(encoding),
                        column.getBytes(encoding), num);
        return qualifie;
    }

    private Table getTable(String tableName) throws IOException {
        if(StringUtils.isNotBlank(tableName)){
            return connection.getTable(TableName.valueOf(tableName));
        }
        return connection.getTable(TABLE_MESSAGE);
    }

    private Table getTable() throws IOException {
        return connection.getTable(TABLE_MESSAGE);
    }

}
